package com.shujia.core

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

object Demo3EventTime {

  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    /*

1,1627971625000
1,1627971626000
1,1627971627000
1,1627971628000
1,1627971630000
1,1627971629000
1,1627971631000
1,1627971632000
1,1627971633000
1,1627971634000
1,1627971635000
1,1627971640000
     */
    /**
      * 使用事件时间
      * 1、需要设置时间模式为事件时间
      * 2、需要告诉flink 哪一个字段是时间字段
      *
      */


    //需要设置时间模式为事件时间
      env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)



    val linesDS: DataStream[String] = env.socketTextStream("master",8888)

    val eventDS: DataStream[Event] = linesDS.map(line =>{
      val splits: Array[String] = line.split(",")
     Event(splits(0),splits(1).toLong)
    })

    eventDS
      //需要告诉flink 哪一个字段是时间字段
        //.assignAscendingTimestamps(event => event.ts)
      .assignTimestampsAndWatermarks(
      //指定数据最大的延迟时间
      new BoundedOutOfOrdernessTimestampExtractor[Event](Time.seconds(5)) {
        //返回事件时间字段
        override def extractTimestamp(element: Event): Long = {
          element.ts
        }
      })
      .map(event =>(event.id,1))
      .keyBy(_._1)
      .timeWindow(Time.seconds(5))
      .sum(1)
      .print()


    env.execute()

  }
  case class Event(id:String,ts: Long)

}
